package com.zx.mes.coolmq.hyl.sender;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zx.mes.coolmq.hyl.util.MqConstants;
import com.zx.mes.coolmq.hyl.util.RabbitMetaMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * RabbitMQ 消息发送者
 * @author hyl
 * @date 2018-3-19
 * @version 1.0.0
 */
@Component
public class RabbitmqSender {

    private static final Logger logger = LoggerFactory.getLogger(RabbitmqSender.class);

    /**
     * 针对字符串
     */
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 这个是amqp协议通用方法
     */
//   @Autowired
    private AmqpTemplate rabbitmqTemmplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public String send(RabbitMetaMessage rabbitMetaMessage) throws JsonProcessingException {

        final String msgId = UUID.randomUUID().toString();

        //缓存
//        redisTemplate.opsForHash().put(MqConstants.MQ_PRODUCER_RETRY_KEY,msgId,rabbitMetaMessage);
        ValueOperations<String, RabbitMetaMessage> operations=redisTemplate.opsForValue();
        operations.set(msgId,rabbitMetaMessage);
        MessagePostProcessor messagePostProcessor=new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setMessageId(msgId);
                //设置消息持久化
                message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                logger.info("message process:{}",message.toString());
                return message;
            }
        };


        ObjectMapper mapper = new ObjectMapper();
        String json = mapper.writeValueAsString(rabbitMetaMessage.getPayload());
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("application/json");
        Message message = new Message(json.getBytes(), messageProperties);
        //发送一个带id(new CorrelationData(msgId))的消息
        try {
            rabbitTemplate.convertAndSend(rabbitMetaMessage.getExchange(),
                    rabbitMetaMessage.getRoutingKey(),
                    message,messagePostProcessor,new CorrelationData(msgId));
            logger.info("发送消息，消息ID:{}",msgId);
            return msgId;
        } catch (AmqpException e) {
            throw new RuntimeException("发送RabbitMQ消息失败！", e);
        }

    }
}
